package cn.chinaunicom.kafka.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Date;

/**
 * @author xuwenyuan
 * @date 2020/2/16 15:54
 * @description
 */

@Service
public class ProductKafka {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(String topic,String message) {
        ListenableFuture future = kafkaTemplate.send(topic, message);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message),
                throwable -> System.out.println("消息发送失败：" + message));
    }
    public void send1(String topic,String message) {
        ListenableFuture future = kafkaTemplate.send(topic,"key demo", message+1);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message),
                throwable -> System.out.println("消息发送失败：" + message));
    }
    public void send2(String topic,String message) {
        ListenableFuture future = kafkaTemplate.send(topic,1,"key demo", message+2);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message),
                throwable -> System.out.println("消息发送失败：" + message));
    }
    public void send3(String topic,String message) {
        Date date = new Date();
        long time = date.getTime();
        ListenableFuture future = kafkaTemplate.send(topic,1,time,"key demo", message+4);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message),
                throwable -> System.out.println("消息发送失败：" + message));
    }

}
